package com.rrd.cannl.consumer;

import java.util.List;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.rrd.cannl.sink.KafkaSink;

public class AccountCannlClient extends AbstractCanalClient{

	public AccountCannlClient(String destination) {
		super(destination);
	}
	public AccountCannlClient(String destination, CanalConnector connector) {
		super(destination,connector);
	}
	/**
	 * 解析rowdata，然后写入到存储
	 * @param rowData
	 * @param type
	 */
	private void rowDataToString(List<Column> columnList,String opType){
		StringBuilder sb = new StringBuilder(500);
		sb.append(opType);
		for(Column c : columnList){
			sb.append(CRT001).append(c.getValue()==null?"":c.getValue());
		}
		if(this.sink==null){
			printColumn(columnList);
		}else{
			
			this.sink.write(columnList.get(2).getValue(),sb.toString().replace("\n", ""));
		}
		
	}
	public static void main(String args[]) {
        String destination = "account";
        String topic ="account.loan";
        // 基于zookeeper动态获取canal server的地址，建立链接，其中一台server发生crash，可以支持failover
        CanalConnector connector = CanalConnectors.newClusterConnector("datanode7.cdh2.app.rrd:2181,datanode8.cdh2.app.rrd:2181,datanode6.cdh2.app.rrd:2181",destination, "", "");
        final AccountCannlClient client = new AccountCannlClient(destination);
        final KafkaSink sink = new KafkaSink(topic);
        client.setConnector(connector);
        client.setSink(sink);
        client.start();

        Runtime.getRuntime().addShutdownHook(new Thread() {

            public void run() {
                try {
                    logger.info("## stop the canal client");
                    client.stop();
                    sink.closeProducer();
                } catch (Throwable e) {
                    logger.warn("##something goes wrong when stopping canal:", e);
                } finally {
                    logger.info("## canal client is down.");
                }
            }
           
        });
    }

}
